package com.urfresh.sp.flume.receive.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Maps;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;

@Service
public class AppEventTrackInfo {

    Logger logger = LoggerFactory.getLogger(AppEventTrackInfo.class);

    @Autowired
    MongoTemplate mongoTemplate;

    /**
     * 接收从kafka传来的点击相关事件的埋点信息
     *
     * @param records
     */
    public void saveAppDataInfo(Iterable<ConsumerRecord<String, String>> records) {
        if (records.iterator().hasNext() == false)
            return;

        for (ConsumerRecord<String, String> r : records) {
            logger.info("app event track=" + r);
            //增加保存时间
            JSONObject obj = JSON.parseObject(r.value());

            obj.put("from", "pc");

            //MongoDB没有自增id，所以需要精确到毫秒级别
//            obj.put("createTime", obj.getString("submitTime"));
            if (!obj.containsKey("createTime"))
                obj.put("createTime", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSSS").format(new Date()));
            save(obj.toJSONString(), "event_track");
        }
    }

    /**
     * 接收从kafka传来的启动相关事件的埋点信息
     *
     * @param records
     */
    public void saveAppSysInfo(Iterable<ConsumerRecord<String, String>> records) {
        if (records.iterator().hasNext() == false)
            return;

//        implantSysInfo.saveImplantInfo(records);

        for (ConsumerRecord<String, String> r : records) {
            logger.info("app sys track=" + r);

            //解析track信息，保存成为json格式
            JSONObject obj = JSON.parseObject(r.value());
            try {
                if (obj.containsKey("trackInfo") && StringUtils.isEmpty(obj.getString("trackInfo")) == false) {
                    String trackInfo = obj.getString("trackInfo");
                    if (trackInfo.contains("&")) {
                        //组成json格式
                        Map<String, String> map = Maps.newConcurrentMap();
                        String[] str = obj.getString("trackInfo").split("&");
                        for (String v : str) {
                            String[] kv = v.split("=");
                            map.put(kv[0], kv[1]);
                        }
                        obj.replace("trackInfo", map);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            if (!obj.containsKey("createTime"))
                obj.put("createTime", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSSS").format(new Date()));
//            obj.put("createTime", obj.getString("submitTime"));
            save(obj.toJSONString(), "sys_track");
        }
    }

    /**
     * 保存android设备的安装列表
     *
     * @param records
     */
    public void saveAppListDataInfo(Iterable<ConsumerRecord<String, String>> records) {
        for (ConsumerRecord<String, String> r : records) {
            logger.info("app event track=" + r);
            //增加保存时间
            JSONObject obj = JSON.parseObject(r.value());
//            obj.put("createTime", obj.getString("submitTime"));
            if (!obj.containsKey("createTime"))
                obj.put("createTime", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSSS").format(new Date()));
            save(obj.toJSONString(), "app_list");
        }
    }

    //保存到mongo
    private void save(String val, String collection) {
        try {
            mongoTemplate.save(val, collection);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
